package com.pxene.dmp.mapper;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.pxene.dmp.common.EncryptUtils;
import com.pxene.dmp.common.KVSender;
import com.pxene.dmp.common.KVUrlGenerator;
import com.pxene.dmp.constant.BaseConstant;

public class SendBusinessCodeMapper extends
		Mapper<LongWritable, Text, Text, LongWritable> {
	
	private Log log = LogFactory.getLog(SendBusinessCodeMapper.class);
	
	private long id;
    private int increment;
    private static final String PREFIX = "b-";
    
    private String date;
    private String apiName;
    
    private Text resultKey = new Text("0");
    private LongWritable resultVal = new LongWritable();
    
    private Map<String, String> params = new HashMap<String, String>();
    
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		// 设置配置文件名
		System.setProperty("config-file", conf.get("config-file"));
		// 0预留给数据大小
		id = context.getTaskAttemptID().getTaskID().getId() + 1;
        increment = context.getConfiguration().getInt("mapreduce.job.maps", 0);
        if (increment == 0) {
            throw new IllegalArgumentException("mapreduce.job.maps is zero");
        }
        // 获取数据日期和api名称
 		date = conf.get("date");
 		apiName = conf.get("api-name");
 		log.info(BaseConstant.LOG_PREFIX + "date=" + date + ",apiName=" + apiName);
	}

	@Override
	protected void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		String[] columns = value.toString().split("\\|");
		if (columns.length != 7) {
			return;
		}
		String imsi = columns[0];
		String bsid = columns[1];
		String starttime = columns[2];
		String name = columns[3];
		String codes = columns[4];
		String province = columns[5];
		String dataType = columns[6];
		// 标签
		StringBuilder tags = new StringBuilder();
		for (String code : codes.split(";")) {
			if ("".equals(code)) {
				continue;
			}
			if (!"JD".equals(name) && !"TMALL".equals(name) && !"TAOBAO".equals(name)) {
				code = EncryptUtils.encryptStr(code);
			}
			tags.append(PREFIX).append(code).append("|");
		}
		int length = tags.length();
		if (length <= 1) {
			return;
		}
		// 发送
		String url = null;
		try {
			url = KVUrlGenerator.getCustUrl(apiName);
		} catch (Exception e) {
			e.printStackTrace();
		}
		params.clear();
		params.put("index", String.valueOf(id));
		params.put("date", date);
		params.put("imsi", imsi);
		params.put("meid", EncryptUtils.randomString(20).toUpperCase() + bsid);
		params.put("name", name);
		params.put("code", tags.substring(0, length-1));
		params.put("time", starttime);
		params.put("province", province);
		params.put("datatype", dataType);
		CountDownLatch latch = new CountDownLatch(1);
		// 异步调用
		KVSender sender = new KVSender(url, params, latch);
		sender.execute();
		// 无论是否成功，都向下进行
		latch.await();
		sender.close();
		
		resultVal.set(id);
		context.write(resultKey, resultVal);
		// id标识递增
		id += increment;
	}
	
}
